import { Inject, Injectable } from "@nestjs/common";
import { Kafka, logLevel, Consumer } from 'kafkajs'
import { LongConnsService } from "./long.conns.service";
import { RedisService } from "./redis.service";
import { MsgService } from "./msg.service";
import { StorageService } from "./storage.service";

let appid2user = { "wx_5l-eC0uimXCDlLMNB99SM": 65 }

@Injectable()
export class PushKafkaService {

    kafkaHost: string = '134.175.224.228:9092'

    constructor(
        private gateWayService: LongConnsService,
        private redisService : RedisService,
        private msgService : MsgService,
        private storageService: StorageService
    ) {

    }

    async onModuleInit() {
        const kafka = new Kafka({
            logLevel: logLevel.INFO,
            brokers: [`${this.kafkaHost}`],
            clientId: 'kafka-push',
        })

        const topic = 'MsgToPush'
        const consumer = kafka.consumer({ groupId: 'push' })

        const run = async () => {
            await consumer.connect()
            await consumer.subscribe({ topic, fromBeginning: true })
            await consumer.run({
                eachMessage: async ({ topic, partition, message }) => {
                    let content = message.value.toString()
                    let data = JSON.parse(content)
                    if (data?.Appid) {
                        // if(data.Data.MsgSeq > 100) 
                        //     return 
                        let conversationID = this.msgService.getConversationIdByMsg(data)

                        let userId : any = await this.storageService.getUserIDByWxid(data.Wxid)
                        //let userId = appid2user[data.Appid]
                        //console.log(userId)

                        if(!userId) return
                            
                        this.gateWayService.pushMessage2User(userId, data)
                    } else {
                        console.log(content)
                    }
                }
            })
        }

        run().catch(e => console.error(`[example/consumer] ${e.message}`, e))
    }

    async onModuleDestroy() {

    }
}